package pers.amos.eshop.storm.bolt;

import com.alibaba.fastjson.JSONArray;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.map.LRUMap;
import org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.utils.Utils;
import pers.amos.eshop.storm.http.HttpClientUtils;
import pers.amos.eshop.storm.zk.ZookeeperSession;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @author amos wong
 * @create 2020-11-23 5:20 下午
 * <p>
 * 商品访问次数统计bolt
 */
@Slf4j
public class ProductCountBolt extends BaseRichBolt {

    private final LRUMap<Long, Long> productCountMap = new LRUMap<>(1000);

    private ZookeeperSession zkSession;
    /**
     * 当前任务的id
     */
    private int taskId;

    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.zkSession = ZookeeperSession.getInstance();
        this.taskId = topologyContext.getThisTaskId();

        new Thread(new ProductCountThread()).start();
        new Thread(new HotProductFindThread()).start();

        initTaskId(topologyContext.getThisTaskId());
    }

    /**
     * 初始化任务id，将所有任务的taskId保存到zk的/taskId-list节点下面
     * 1.将自己的taskId写入一个zookeeper node中，形成taskId的列表
     * 2.然后每次都将自己的热门商品列表，写入自己的taskId对应的zookeeper节点
     * 3.然后这样的话，并行的预热程序才能从第一步中知道，有哪些taskId
     * 4.然后并行预热程序根据每个taskId去获取一个锁，然后再从对应的zNode中拿到热门商品列表
     *
     * @param taskId 当前任务的Id
     */
    public void initTaskId(int taskId) {

        zkSession.acquireDistributeLock("/taskId-list-lock");

        // 先去创建该节点，如果存在则不创建 防止出现该节点未创建的情况
        zkSession.createNode("/taskId-list");

        String taskIdList = zkSession.getNodeData("/taskId-list"); // 获取该节点下的值
        log.info("【ProductCountBolt获取到taskId list】taskIdList={}", taskIdList);

        if (!StringUtils.isEmpty(taskIdList)) {
            taskIdList += "," + taskId;
        } else {
            taskIdList += taskId;
        }

        zkSession.setNodeData("/taskId-list", taskIdList);
        log.info("【ProductCountBolt设置taskId list】taskIdList={}", taskIdList);

        zkSession.releaseDistributeLock("/taskId-list-lock");
    }

    @Override
    public void execute(Tuple tuple) {
        Long productId = tuple.getLongByField("productId");

        log.info("【ProductCountBolt接收到一个商品id】 productId={}", productId);

        Long count = productCountMap.get(productId);
        if (count == null) {
            count = 0L;
        }
        count++;
        productCountMap.put(productId, count);

        log.info("【ProductCountBolt完成商品访问次数统计】productId={}, count= {}", productId, count);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    /**
     * 工作线程 用于计算出topN热点商品
     * 计算出TopN热点商品是为了缓存预热
     */
    private class ProductCountThread implements Runnable {

        public void run() {
            List<Map.Entry<Long, Long>> topProductList = new ArrayList<>();
            List<Long> productIdList = new ArrayList<>();

            while (true) {
                try {
                    topProductList.clear();
                    productIdList.clear();

                    int topN = 3;

                    if (productCountMap.size() == 0) {
                        Utils.sleep(100);
                        continue;
                    }

                    log.info("【ProductCountThread打印productCountMap的长度】size={}", productCountMap.size());

                    for (Map.Entry<Long, Long> productCountEntry : productCountMap.entrySet()) {
                        if (topProductList.size() == 0) {
                            topProductList.add(productCountEntry);
                        } else {
                            // 比较大小
                            boolean bigger = false;

                            for (int i = 0; i < topProductList.size(); i++) {
                                Map.Entry<Long, Long> topProductCountEntry = topProductList.get(i);

                                if (productCountEntry.getValue() > topProductCountEntry.getValue()) {
                                    int lastIndex = topProductList.size() < topN ? topProductList.size() - 1 : topN - 2;
                                    for (int j = lastIndex; j >= i; j--) {
                                        if (j + 1 == topProductList.size()) {
                                            topProductList.add(null);
                                        }
                                        topProductList.set(j + 1, topProductList.get(j));
                                    }
                                    topProductList.set(i, productCountEntry);
                                    bigger = true;
                                    break;
                                }
                            }

                            if (!bigger) {
                                if (topProductList.size() < topN) {
                                    topProductList.add(productCountEntry);
                                }
                            }
                        }
                    }

                    for (Map.Entry<Long, Long> topProductEntry : topProductList) {
                        productIdList.add(topProductEntry.getKey());
                    }

                    // 每次循环之后需要将topNProductList放入到taskId的节点下面
                    String topProductJson = JSONArray.toJSONString(productIdList);
                    zkSession.createNode("/task-hot-product-list-" + taskId); // 先创建该节点 防止不存在
                    zkSession.setNodeData("/task-hot-product-list-" + taskId, topProductJson);

                    log.info("【ProductCountThread计算出一份top3热门商品列表】zk path={}, topProductListJSON={}", "/task-hot-product-list-" + taskId, topProductJson);

                    Utils.sleep(5000); // 休息5s
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 用于统计出热点商品的线程 为了解决热点商品问题
     * 统计热点商品是为了防止热点商品访问量过大导致nginx宕机
     * 热点商品：访问次数 >= N * AVG(95%低访问量商品的访问次数）
     */
    private class HotProductFindThread implements Runnable {

        /**
         * 1、将LRUMap中的数据按照访问次数，进行全局的排序
         * 2、计算95%的商品的访问次数的平均值
         * 3、遍历排序后的商品访问次数，从最大的开始
         * 4、如果某个商品比如它的访问量是平均值的10倍，就认为是缓存的热点
         */
        @Override
        public void run() {
            List<Map.Entry<Long, Long>> productCountList = new ArrayList<>();
            List<Long> hotProductIdList = new ArrayList<>();
            // 记录上次的热点商品列表 用于与此次的热点商品比较 取消掉不再是热点的商品
            List<Long> lastTimeHotProductIdList = new ArrayList<>();

            while (true) {
                try {
                    productCountList.clear();
                    hotProductIdList.clear();

                    if (productCountMap.size() == 0) {
                        Utils.sleep(100);
                        continue;
                    }

                    log.info("【HotProductFindThread打印productCountMap的长度】size={}", productCountMap.size());

                    // 1、先做全局的排序

                    for (Map.Entry<Long, Long> productCountEntry : productCountMap.entrySet()) {
                        if (productCountList.size() == 0) {
                            productCountList.add(productCountEntry);
                        } else {
                            boolean bigger = false;

                            for (int i = 0; i < productCountList.size(); i++) {
                                Map.Entry<Long, Long> topNProductCountEntry = productCountList.get(i);

                                if (productCountEntry.getValue() > topNProductCountEntry.getValue()) {
                                    int lastIndex = productCountList.size() < productCountMap.size() ? productCountList.size() - 1 : productCountMap.size() - 2;
                                    for (int j = lastIndex; j >= i; j--) {
                                        if (j + 1 == productCountList.size()) {
                                            productCountList.add(null);
                                        }
                                        productCountList.set(j + 1, productCountList.get(j));
                                    }
                                    productCountList.set(i, productCountEntry);
                                    bigger = true;
                                    break;
                                }
                            }

                            if (!bigger) {
                                if (productCountList.size() < productCountMap.size()) {
                                    productCountList.add(productCountEntry);
                                }
                            }
                        }
                    }

                    log.info("【HotProductFindThread全局排序后的结果】productCountList={}", productCountList);

                    // 2、计算出95%的商品的访问次数的平均值
                    int calculateCount = (int) Math.floor(productCountList.size() * 0.95);

                    Long totalCount = 0L;
                    for (int i = productCountList.size() - 1; i >= productCountList.size() - calculateCount; i--) {
                        totalCount += productCountList.get(i).getValue();
                    }

                    // Math.floor()并不是四舍五入 Math.floor(0.95)=0.0 所以如果上面productCountList=1
                    // 那么calculateCount=0 因此下面做除法运算时就会报算数异常 因此需要添加此逻辑避免错误
                    if (calculateCount == 0) {
                        calculateCount = 1;
                    }
                    long avgCount = totalCount / calculateCount;

                    log.info("【HotProductFindThread计算出95%的商品的访问次数平均值】avgCount={}", avgCount);

                    // 3、从第一个元素开始遍历，判断是否是平均值得10倍
                    for (Map.Entry<Long, Long> productCountEntry : productCountList) {
                        if (productCountEntry.getValue() >= 10 * avgCount) {
                            log.info("【HotProductFindThread发现一个热点】productCountEntry={}", productCountEntry);
                            hotProductIdList.add(productCountEntry.getKey());

                            // 4.推送数据到nginx
                            // 如果上次已经推送过了 这次就不推送了 避免重复查询redis带来的性能消耗
                            if (!lastTimeHotProductIdList.contains(productCountEntry.getKey())) {
                                pushProductInfo2Nginx(productCountEntry.getKey());
                            }
                        }
                    }

                    // 4.实时感知热点数据的消失
                    if (lastTimeHotProductIdList.size() == 0) {
                        if (hotProductIdList.size() > 0) {
                            lastTimeHotProductIdList.addAll(hotProductIdList);
                            log.info("【HotProductFindThread保存上次热点数据】lastTimeHotProductIdList={}", lastTimeHotProductIdList);
                        }
                    } else {
                        for (Long productId : lastTimeHotProductIdList) {
                            if (!hotProductIdList.contains(productId)) {
                                log.info("【HotProductFindThread发现一个热点消失了】productId={}", productId);
                                // 发送一个http请求给到流量分发的nginx中，取消热点缓存的标识(设置为false即可）
                                String cancelHotProductUrl = "http://127.0.0.1:10000/cancel_hot?productId=" + productId;
                                HttpClientUtils.sendGetRequest(cancelHotProductUrl);
                            }
                        }

                        if (hotProductIdList.size() > 0) {
                            lastTimeHotProductIdList.clear();
                            lastTimeHotProductIdList.addAll(hotProductIdList);
                            log.info("【HotProductFindThread保存上次热点数据】lastTimeHotProductIdList={}", lastTimeHotProductIdList);
                        } else {
                            // 本次没有热点商品 上次的热点商品list需要全部清空
                            lastTimeHotProductIdList.clear();
                        }
                    }


                    Utils.sleep(5000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }

        /**
         * 1.将热点商品id推送到流量分发服务器
         * 2.将热点商品的完整信息 推送到所有的应用服务器
         *
         * @param productId
         */
        private void pushProductInfo2Nginx(Long productId) {
            // 将缓存热点反向推送到流量分发的nginx中
            String distributeNginxUrl = "http://127.0.0.1:10000/hot?productId=" + productId;
            HttpClientUtils.sendGetRequest(distributeNginxUrl);

            // 将缓存热点，那个商品对应的完整的缓存数据，发送请求到缓存服务去获取，反向推送到所有的后端应用nginx服务器上去
            String cacheServiceUrl = "http://127.0.0.1:8081/getProductInfo?productId=" + productId;
            String response = HttpClientUtils.sendGetRequest(cacheServiceUrl);

            String[] appNginxUrls = new String[]{
                    "http://127.0.0.1:10010/hot?productId=" + productId + "&productInfo=" + response,
                    "http://127.0.0.1:10086/hot?productId=" + productId + "&productInfo=" + response
            };

            // 将查询出的商品完整缓存信息推送到应用nginx上
            for (String appNginxUrl : appNginxUrls) {
                HttpClientUtils.sendGetRequest(appNginxUrl);
            }
        }
    }

}
